package com.cloud.common.akka;

import akka.actor.ActorRef;
import akka.actor.ActorSystem;
import akka.actor.Props;
import org.springframework.boot.autoconfigure.condition.ConditionalOnClass;
import org.springframework.stereotype.Component;

import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;

@Component
@ConditionalOnClass(ActorSystem.class)
public class ActorManager {

    private Map<Class, ActorRef> actorRefMap = new ConcurrentHashMap<>();

    private ActorSystem actorSystem;

    public void tell(Class actor, ActorMessage message) {
        get(actor).tell(message, ActorRef.noSender());
    }

    private ActorRef get(Class clazz) {
        ActorRef actorRef = actorRefMap.get(clazz);
        if (actorRef == null) {
            actorRef = getActorSystem().actorOf(Props.create(ActorProducer.class, clazz));
            actorRefMap.put(clazz, actorRef);
        }
        return actorRef;
    }

    private synchronized ActorSystem getActorSystem() {
        if (actorSystem == null) {
            actorSystem = ActorSystem.create("sys");
        }
        return actorSystem;
    }

}
